package com.atguigu.flink.state;

import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/23
 *
 *
 *  unionState: 是ListState的变种。
 *          ListState： 在初始化的时候，状态是均匀分配。
 *                 存储的状态有a,b,c,d
 *                      重启后有2个并行度:
 *                              task1:  a , c
 *                              task2:  b , d
 *                       重启后有4个并行度:
 *                              task1 : a
 *                              task2 : b
 *                              task3 : c
 *                              task4 : d
 *
 *           UnionState:  在初始化的时候，状态是 先union（合并），再分配。每个并行度获取的状态是全部状态
 *                       存储的状态有a,b,c,d
 *  *                      重启后有2个并行度:
 *  *                              task1: a,b,c,d
 *  *                              task2:  a,b,c,d
 *  *                       重启后有4个并行度:
 *  *                              task1 : a,b,c,d
 *  *                              task2 : a,b,c,d
 *  *                              task3 : a,b,c,d
 *  *                              task4 : a,b,c,d
 *
 */
public class Demo2_UnionState
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);


        env.setParallelism(2);

        env.enableCheckpointing(500);
        //存到文件系统
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:9820/ck");

        //使用rocksdb,额外引入依赖
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        //读取数据
        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888);

        //把每次读到的数据存入到一个List集合中
        ds.flatMap(new MyFlatMap())
          .addSink(new SinkFunction<String>()
          {
              @Override
              public void invoke(String value, Context context) throws Exception {
                  if (value.contains("x")){
                      throw new RuntimeException("出异常了...");
                  }
                  System.out.println(value);
              }
          });

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }

    private static class  MyFlatMap implements  FlatMapFunction<String,String>, CheckpointedFunction
    {

        //不要声明普通的类型的 属性，而应该用Flink提供的 ManagedState(属性，自动备份)
       // private List<String> strs=new ArrayList<>();
        // add: 加一个
        // addAll: 加一个List
        // clear: 清空
        // update ： 等价于 先 clear再addAll
        // get：  获取里面的元素
        private ListState<String> state;

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {

            //strs.add(value);
            state.add(value);

            out.collect(MyUtil.parseList(state.get()).toString());

        }

        //快照状态，备份  周期型执行
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("MyFlatMap.snapshotState");
        }

        //初始化状态  恢复  只在Task'启动时执行一次
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyFlatMap.initializeState");

            //声明状态, 从备份中读取状态
            state = context.getOperatorStateStore().getUnionListState(new ListStateDescriptor<String>("list", String.class));

        }
    }
}
